package cn.dfun.sample.flink.apitest
import java.util.Properties

import org.apache.flink.api.common.serialization.{SimpleStringEncoder, SimpleStringSchema}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}
import org.apache.flink.streaming.api.scala._

/**
  * kafka sink
  * 可以用flink做etl
  * 相关脚本:
  * 启动消费者
  * ./bin/kafka-console-consumer.sh --bootstrap-server node-01:9092 --topic sinktest
  * 启动生产者
  * ./bin/kafka-console-producer.sh --broker-list node-01:9092 --topic sensor
  */
object KafkaSinkTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
    val inputStream= env.readTextFile(inputPath)
    // 包装成样例类
    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble).toString
      })

    // 从kafka读取数据
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "node-01:9092")
    properties.setProperty("group.id", "consumer-group")

    // kafka sink
    val stream = env.addSource(new FlinkKafkaConsumer011[String]("sensor",
      new SimpleStringSchema(), properties))

    stream.addSink(new FlinkKafkaProducer011[String]("node-01:9092", "sinktest", new SimpleStringSchema()) )
    env.execute("kafka sink test")
  }
}
